package com.juneyao.bigdata.service.impl;

import com.alibaba.fastjson.JSON;
import com.juneyao.bigdata.common.ConstantNV;
import com.juneyao.bigdata.domain.*;
import com.juneyao.bigdata.service.ElasticSearchService;
import com.juneyao.bigdata.util.FieldKeyWordJoin;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author quanbiao
 * @date 2018-06-28
 * es查询实现类
 */
@Service("elasticSearchService")
public class ElasticSearchServiceImpl implements ElasticSearchService{

    private Logger logger = LoggerFactory.getLogger("ElasticSearchServiceImpl");

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    private RestTemplate restTemplate;

    @Value("${crowd.label.url}")
    private String crowdUrl;

    /**
     * 核心线程
     */
    private static final int CORE_POOL_SIZE = 10;

    /**
     * 最大线程
     */
    private static final int MAX_NUM_POOL_SIZE = 20;

    /**
     * 线程空闲存活时间
     */
    private static final long THREAD_POOL_KEEP_ALIVE_TIME = 10000L;

    /**
     * 线程池
     */
    private static final ThreadPoolExecutor elasticSearchThreadPool;

    @Value("${jx.user.inter.topic}")
    private String kafkaTopic;

    /**
     * 初始化线程池
     */
    static {
        elasticSearchThreadPool = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_NUM_POOL_SIZE,
                THREAD_POOL_KEEP_ALIVE_TIME,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>());
        elasticSearchThreadPool.allowCoreThreadTimeOut(true);
    }
    /**
     * 根据插入的参数，获取用户的id，作为rowKey去Hbase检索
     * @param groupPeopleVO
     * @return
     */
    @Override
    public PageValueCount queryUserIdsForPage(GroupPeopleVO groupPeopleVO) {

        logger.info("groupPeopleVO =" + groupPeopleVO.toString());

        //返回数据
        PageValueCount pageValueCount = new PageValueCount();

        //返回值，用户id
        List<String> userIdArr = new ArrayList<>();

        //分页信息-页码
        Integer pageNum = groupPeopleVO.getPageNum();
        //分页信息-条数
        Integer pageSize = groupPeopleVO.getPageSize();

        //构建es分页request，es的分页是从0开始的，索引使用页码减一
        PageRequest pageRequest = PageRequest.of(pageNum - 1, pageSize);

        //根据条件获取es查询build
        BoolQueryBuilder conQueryUltimate = getBoolQueryBuilder(groupPeopleVO.getCrowdId(), ConstantNV.MEMBER_FLAG_NO,
                ConstantNV.BLOCK_FLAG_NO, ConstantNV.WEI_CHAT_FLAG_NO);

        //构建查询build
        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                //索引名称
                .withIndices(ConstantNV.ES_INDEX_NAME)
                //索引类型
                .withTypes(ConstantNV.ES_INDEX_TYPE)
                //分页信息
                .withPageable(pageRequest)
                //查询字段
                .withFields(ConstantNV.ES_DOCUMENT_ID)
                //组合查询build
                .withFilter(conQueryUltimate).build();
        AggregatedPage<UserDetail> userDetailAggregatedPage = elasticsearchTemplate.queryForPage(searchQuery, UserDetail.class);
        //获取分页数据
        List<UserDetail> content = userDetailAggregatedPage.getContent();
        for (UserDetail userDetail : content) {
            userIdArr.add(userDetail.getId());
        }
        //分页展示的用户id
        pageValueCount.setUserIds(userIdArr);
        //总条数赋值
        pageValueCount.setCount(userDetailAggregatedPage.getTotalElements());
        return pageValueCount;
    }

    /**
     * 获取集合人数
     * @param peopleGroupId
     * @return
     */
    @Override
    public PeopleGroup peopleGroupSum(PeopleGroupId peopleGroupId) {

        //集合人数实体
        PeopleGroup peopleGroup = new PeopleGroup();
        //每个集合对应的人数
        Map<String, Long> userCountEachGroup = new HashMap<>(16);
        //获取集合id
        String[] groupIdArr = peopleGroupId.getIds().split(",");

        //初始化计数器
        CountDownLatch countDownLatch = new CountDownLatch(groupIdArr.length + 3);

        //异步获取，值先使用future保存
        Map<String, Future<Long>> otherLabelMap = new HashMap<>(16);

        //每个集合对应的人数--异步获取
        Map<String, Future<Long>> userCountEachGroupFuture = new HashMap<>(16);
        //循环每个集合，获取每个集合的人数
        for (String groupId : groupIdArr) {
            BoolQueryBuilder boolQueryBuilder = getBoolQueryBuilder(groupId, ConstantNV.MEMBER_FLAG_NO,
                    ConstantNV.BLOCK_FLAG_NO, ConstantNV.WEI_CHAT_FLAG_NO);
            //构建查询build
            SearchQuery searchQueryBuilder = new NativeSearchQueryBuilder()
                    //索引名称
                    .withIndices(ConstantNV.ES_INDEX_NAME)
                    //索引类型
                    .withTypes(ConstantNV.ES_INDEX_TYPE)
                    .withFilter(boolQueryBuilder).build();
            try {
                Future<Long> userCountFuture = elasticSearchThreadPool.submit(new ElasticRunnable(elasticsearchTemplate, searchQueryBuilder, countDownLatch));
                userCountEachGroupFuture.put(groupId, userCountFuture);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        //去重后的人数
        BoolQueryBuilder someGroupBoolQuery = getBoolQueryBuilder(peopleGroupId.getIds(), ConstantNV.MEMBER_FLAG_NO,
                ConstantNV.BLOCK_FLAG_NO, ConstantNV.WEI_CHAT_FLAG_NO);
        //构建查询build
        SearchQuery searchSomeGroupBoolBuilder = new NativeSearchQueryBuilder()
                //索引名称
                .withIndices(ConstantNV.ES_INDEX_NAME)
                //索引类型
                .withTypes(ConstantNV.ES_INDEX_TYPE)
                .withFilter(someGroupBoolQuery).build();
        try {
            Future<Long> userUniqueCountFuture = elasticSearchThreadPool.submit(new ElasticRunnable(elasticsearchTemplate, searchSomeGroupBoolBuilder, countDownLatch));
            otherLabelMap.put("distinctSum", userUniqueCountFuture);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //去除黑名单 + 是会员
        BoolQueryBuilder memberBoolQueryBuilder = getBoolQueryBuilder(peopleGroupId.getIds(), ConstantNV.MEMBER_FLAG_YES,
                ConstantNV.BLOCK_FLAG_YES,ConstantNV.WEI_CHAT_FLAG_NO);
        //构建查询build
        SearchQuery searchMemGroupBoolBuilder = new NativeSearchQueryBuilder()
                //索引名称
                .withIndices(ConstantNV.ES_INDEX_NAME)
                //索引类型
                .withTypes(ConstantNV.ES_INDEX_TYPE)
                .withFilter(memberBoolQueryBuilder).build();
        //获取会员和去掉黑名单的人数
        try {
            Future<Long> memberCountFuture = elasticSearchThreadPool.submit(new ElasticRunnable(elasticsearchTemplate, searchMemGroupBoolBuilder, countDownLatch));
            otherLabelMap.put("memberSum", memberCountFuture);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //微信可触达人数
        BoolQueryBuilder weiChatBoolQueryBuilder = getBoolQueryBuilder(peopleGroupId.getIds(), ConstantNV.MEMBER_FLAG_YES,
                ConstantNV.BLOCK_FLAG_YES,ConstantNV.WEI_CHAT_FLAG_YES);
        //构建查询build
        SearchQuery searchWeiChatGroupBoolBuilder = new NativeSearchQueryBuilder()
                //索引名称
                .withIndices(ConstantNV.ES_INDEX_NAME)
                //索引类型
                .withTypes(ConstantNV.ES_INDEX_TYPE)
                .withFilter(weiChatBoolQueryBuilder).build();
        try {
            Future<Long> weiChatCountFuture = elasticSearchThreadPool.submit(new ElasticRunnable(elasticsearchTemplate, searchWeiChatGroupBoolBuilder, countDownLatch));
            otherLabelMap.put("weiChatSum", weiChatCountFuture);
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            //线程同步，超过30秒，继续执行
            countDownLatch.await(30, TimeUnit.SECONDS);
            //去重后人数
            Long distinctSum = otherLabelMap.get("distinctSum").get();
            //会员(去除黑名单)
            Long memberSum = otherLabelMap.get("memberSum").get();
            //微信可触达人数
            Long weiChatSum = otherLabelMap.get("weiChatSum").get();
            peopleGroup.setWechatSum(weiChatSum);
            //去重后人数
            peopleGroup.setDistinctSum(distinctSum);
            //会员(去除黑名单)
            peopleGroup.setDistinctBlacklistSum(memberSum);
            //app可触达人数
            peopleGroup.setAppSum(memberSum);
            //短信可触达人数
            peopleGroup.setSmsSum(memberSum);
            //二次过滤
            peopleGroup.setDistinctSecondaryFilterSum(memberSum);
            //每个集合对应的人数
            for (Map.Entry<String, Future<Long>> userGroupCount : userCountEachGroupFuture.entrySet()) {
                userCountEachGroup.put(userGroupCount.getKey(), userGroupCount.getValue().get());
            }
            peopleGroup.setPeoplegroupSum(userCountEachGroup);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }catch (ExecutionException e) {
            e.printStackTrace();
        }
        return peopleGroup;
    }

    /**
     * 根据多个集合，生成查询build
     * @param crowdIdStr
     * @param memberFlag 会员标识
     * @param weiChatFlag 微信标识
     * @return
     */
    @Override
    public BoolQueryBuilder getBoolQueryBuilder(String crowdIdStr, String memberFlag, String blockFlag, String weiChatFlag){
        //拆分
        String[] crowdIdArr = crowdIdStr.split(",");

        //总的：条件组合查询
        //前台传过来多个集合，每个集合内是交集，多个集合是并集
        BoolQueryBuilder conQueryUltimate = QueryBuilders.boolQuery();

        //循环集合id，通过集合id获取标签
        for (String crowdId : crowdIdArr) {
            //每个集合的查询build
            BoolQueryBuilder condQueryOneSet = QueryBuilders.boolQuery();

            //循环每个集合，获取标签
            ResultParam forObject = restTemplate.getForObject(crowdUrl + crowdId, ResultParam.class);
            SetGroupVo setGroupVo = JSON.parseObject(JSON.toJSONString(forObject.getData()), SetGroupVo.class);
            //获取子集合list，去掉父id为0的
            List<ListCmpPeoplegroupVo> sonSets = setGroupVo.getListCmpPeoplegroupVo()
                    .stream().filter(item -> !item.getPid().equals("0")).collect(Collectors.toList());
            //子集合循环：集合1，集合2
            for (ListCmpPeoplegroupVo sonSet : sonSets) {
                //获取每个子集合的标签
                List<LabelVo> sonSetLabels = sonSet.getListCmpMarklabel();
                //同一个子集合的相同的标签放在一起，取交集
                //map中的key是标签的code，value是标签枚举的list
                Map<String, List<String>> labelGroup = new HashMap<>(16);
                for (LabelVo sonSetLabel : sonSetLabels) {
                    //如果map中不存在该标签id，就加进去
                    if(!labelGroup.keySet().contains(sonSetLabel.getLabelCode())){
                        List<String> temp = new ArrayList();
                        temp.add(sonSetLabel.getLabelEnum());
                        labelGroup.put(sonSetLabel.getLabelCode(), temp);
                    }else{
                        //如果存在，先根据labelCode获取相应的枚举值
                        String mapKeyCode = sonSetLabel.getLabelCode();
                        List<String> labelEnumArr = labelGroup.get(mapKeyCode);
                        labelEnumArr.add(sonSetLabel.getLabelEnum());
                        labelGroup.put(sonSetLabel.getLabelCode(), labelEnumArr);
                    }
                }
                //每个子集合的查询build
                BoolQueryBuilder condQuerySonSet = QueryBuilders.boolQuery();
                //labelGroup根据标签code进行分组
                for (Map.Entry<String, List<String>> labelEnum : labelGroup.entrySet()) {
                    //每个标签对应的多个值
                    TermsQueryBuilder labelEnumsBuild = QueryBuilders
                            .termsQuery(FieldKeyWordJoin.getFullKeyWord(labelEnum.getKey()), labelEnum.getValue());
                    //每个子集合的不同标签取交集，使用filter
                    condQuerySonSet.filter(labelEnumsBuild);
                    //如果是查询会员标识，只查询会员(顺便去重黑名单)
                    if(StringUtils.equals(memberFlag, ConstantNV.MEMBER_FLAG_YES)){
                        //会员查询
                        TermQueryBuilder memberQueryBuild = QueryBuilders
                                .termQuery(FieldKeyWordJoin.getFullKeyWord(ConstantNV.MEMBER_FLAG), ConstantNV.FLAG_YES_VALUE);

                        //只查询会员的
                        condQuerySonSet.filter(memberQueryBuild);
                    }
                    //去掉黑名单
                    if(StringUtils.equals(blockFlag, ConstantNV.BLOCK_FLAG_YES)){
                        //去除黑名单
                        TermQueryBuilder blockQueryBuild = QueryBuilders
                                .termQuery(FieldKeyWordJoin.getFullKeyWord(ConstantNV.BLOCK_FLAG), ConstantNV.FLAG_YES_VALUE);
                        //mustNot是去除
                        condQuerySonSet.mustNot(blockQueryBuild);
                    }
                    //如果是微信标识
                    if(StringUtils.equals(weiChatFlag, ConstantNV.WEI_CHAT_FLAG_YES)){
                        //微信查询
                        TermQueryBuilder weiChatQueryBuild = QueryBuilders
                                .termQuery(FieldKeyWordJoin.getFullKeyWord(ConstantNV.WEI_CHAT_FLAG), ConstantNV.FLAG_YES_VALUE);
                        condQuerySonSet.filter(weiChatQueryBuild);
                    }
                }
                //一个集合中，不同的子集合使用或的关系
                condQueryOneSet.should(condQuerySonSet);
            }
            //对于每个集合，使用或的关系
            conQueryUltimate.should(condQueryOneSet);
        }
        return conQueryUltimate;
    }

    /**
     * 根据插叙条件获取用户id集合，用于发送
     * @param boolQueryBuilder
     * @return
     */
    @Override
    public List<String> getUserIdsForAction(BoolQueryBuilder boolQueryBuilder, int pageNum, int pageSize) {
        //构建es分页request，es的分页是从0开始的，索引使用页码减一
        PageRequest pageRequest = PageRequest.of(pageNum, pageSize);

        //构建查询build
        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                //索引名称
                .withIndices(ConstantNV.ES_INDEX_NAME)
                //索引类型
                .withTypes(ConstantNV.ES_INDEX_TYPE)
                //分页信息
                .withPageable(pageRequest)
                //查询字段
                .withFields(ConstantNV.ES_DOCUMENT_ID)
                //组合查询build
                .withFilter(boolQueryBuilder).build();
        return elasticsearchTemplate.queryForIds(searchQuery);
    }

    /**
     * 根据查询条件获取总记录数
     *
     * @param searchQuery
     * @return
     */
    @Override
    public long searchCount(SearchQuery searchQuery) {
        return elasticsearchTemplate.count(searchQuery);
    }
}
